package service

import (
	"context"
	"encoding/json"
	"github.com/pkg/errors"
)

type CanalBody struct {
	Data      []interface{} `json:"data"`     // 实际数据
	Database  string        `json:"database"` // 数据库或schema
	ES        int64         `json:"es"`       // sql的执行时间
	Gtid      string        `json:"gtid"`
	Id        int64         `json:"id"`        // canal的自增id,不同数据可能重复,不能用于去重
	IsDdl     bool          `json:"isDdl"`     // true/false 是否为ddl
	MysqlType interface{}   `json:"mysqlType"` // 具体的字段类型
	Old       []interface{} `json:"old"`       // 老数据
	PkNames   []string      `json:"pkNames"`
	Sql       string        `json:"sql"`
	SqlType   interface{}   `json:"sqlType"`
	Table     string        `json:"table"` // 表名
	Ts        int64         `json:"ts"`    // canal解析时间
	Type      string        `json:"type"`  // INSERT,UPDATE,DELETE, 如果是ddl,那么值为ALTER
}

var typeFunc = map[bool]func(ctx context.Context, body CanalBody) error{
	false: dml,
	true:  ddl,
}

func parseAnalysisMessage(ctx context.Context, msg CanalBody) error {
	f, ok := typeFunc[msg.IsDdl]
	if !ok {
		return nil
	}
	return f(ctx, msg)
}

func dml(ctx context.Context, body CanalBody) error {
	if body.Database != "wikitrade_business" {
		return errors.New("insert database != wikitrade_business")
	}
	if len(body.Data) < 1 {
		return nil
	}
	data := make([]map[string]interface{}, 0, len(body.Data))
	for _, v := range body.Data {
		dataByte, err := json.Marshal(v)
		if err != nil {
			return errors.Wrap(err, "insert json marshal err:%v")
		}
		dataMap := make(map[string]interface{})
		err = json.Unmarshal(dataByte, &dataMap)
		if err != nil {
			return errors.Wrap(err, "insert json unmarshal err:%v")
		}
		data = append(data, dataMap)
	}
	if body.Type == "INSERT" {
		return AnalysisServer.InsertMessage(ctx, body.Table, data...)
	}
	if body.Type == "UPDATE" {
		return AnalysisServer.UpdateMessage(ctx, body.Table, data...)
	}
	if body.Type == "DELETE" {
		return AnalysisServer.DeleteMessage(ctx, body.Table, data...)
	}
	return nil
}

func ddl(ctx context.Context, body CanalBody) error {
	if body.Database != "wikitrade_business" {
		return errors.New("insert database != wikitrade_business")
	}
	return nil
}
